package AnnualIncome;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class AnnualIncomeReducer extends Reducer<FloatWritable, IntWritable, Text, FloatWritable> {
    private FloatWritable valueOut0 = new FloatWritable();
    private FloatWritable valueOut1 = new FloatWritable();
    private FloatWritable valueOut2 = new FloatWritable();
    private FloatWritable valueOut3 = new FloatWritable();
    private FloatWritable valueOut4 = new FloatWritable();

    int sum0 = 0;
    int sum1 = 0;
    int sum2 = 0;
    int sum3= 0;
    int sum4 = 0;

    @Override
    protected void reduce(FloatWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        if (key.get() == 1) {

            for (IntWritable value : values) {
                sum0 += value.get();
            }
        } else if (key.get() == 2) {
            for (IntWritable value : values) {
                sum1 += value.get();
            }
        } else if (key.get() == 3) {
            for (IntWritable value : values) {
                sum2 += value.get();
            }
        } else if (key.get() == 4) {
            for (IntWritable value : values) {
                sum3 += value.get();
            }
        } else {
            for (IntWritable value : values) {
                sum4 += value.get();
            }
        }
    }


    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        valueOut0.set(sum0);
        valueOut1.set(sum1);
        valueOut2.set(sum2);
        valueOut3.set(sum3);
        valueOut4.set(sum4);

        context.write(new Text("0~30000"), valueOut0);
        context.write(new Text("30001~60000"), valueOut1);
        context.write(new Text("60001~90000"), valueOut2);
        context.write(new Text("90001~120000"), valueOut3);
        context.write(new Text("120001~120000+"), valueOut4);
    }
}
